//============================================================================
// Name        : testzmq.cpp
// Author      : 
// Version     :
// Copyright   : Your copyright notice
// Description : Hello World in C++, Ansi-style
//============================================================================

#include <iostream>
#include <boost/make_shared.hpp>
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/progress.hpp>

#include <string>
#include <fstream>
#include "zmq/DataQueue.h"

boost::shared_ptr<core::zmq::DataSend> m_pSend;
using namespace std;

const char testQ[] = { "tcp://127.0.0.1:7788" }; //{"ipc:///tmp/test.ipc"};
const int ITEMS = 25;
const int QUEUE_SIZE = 10000;

void testSend() {
	m_pSend = boost::make_shared<core::zmq::DataSend>(testQ, QUEUE_SIZE); //"tcp://127.0.0.1:12345");

	RERUN: {
		boost::progress_timer pt;
		string s = "select * from abc where a=1 and b=1 and c=2";
		string p = "sql Plus dev progrmae c:\\progt文件 \\sdddd";
		int len;
		//在发送+接收水位的条数满了以后
		//异步非阻塞模式： drop掉
		//同步阻塞模式 :  等待
#if 1//异步非阻塞模式
		int flags = ZMQ_SNDMORE|ZMQ_DONTWAIT;
		int flagsDone = ZMQ_DONTWAIT;
#else//同步阻塞模式
		int flags = ZMQ_SNDMORE;
		int flagsDone = 0;
#endif

		for (int i = 1; i < 20 * 10000; i++) {
			//cout << "send..." << i << std::endl;
			for (int j = 0; j < ITEMS; j++) {
				len = (int) s.length();
				m_pSend->sendI(len, flags);
				m_pSend->send(s, flags);

				len = (int) p.length();
				m_pSend->sendI(len, flags);
				m_pSend->send(p, flags);
			}

			//m_pSend->sendI(i, 0); ////ZMQ_DONTWAIT);
			m_pSend->sendI(i, flagsDone);
			//if (i % 1000 == 0)
			cout << "send " << i << std::endl;
		}
	}

	cout << "zmq end, enter 0 for rerun" << endl;
	int i;
	cin >> i;
	if (i == 0) {
		goto RERUN;
	}
}

class DataRecvTest: public core::zmq::DataRecvThread {
public:
	DataRecvTest(std::string q, int water) :
			core::zmq::DataRecvThread(q, water) {
		count = 0;
	}
	virtual ~DataRecvTest() {
	}

	int count;
	int i;
	std::string text;
protected:
	virtual void oneWork() {
		for (int j = 0; j < ITEMS; j++) {
			this->recvI(i);
			this->recv(text);
			this->recvI(i);
			this->recv(text);
		}
		this->recvI(i);

		count++;
		if (count % QUEUE_SIZE == 0)
			cout << "recv:" << count << "::" << i << endl;
	}
};

void testRecv() {
	DataRecvTest* recv = new DataRecvTest(testQ, QUEUE_SIZE);
	cout << "start recv" << endl;
	recv->Start();

	cout << "。。。" << endl;
	int i;
	cin >> i;
}

/*
 * 		./testzmq r //RECV mode[pure c++]
 *  	./testzmq //Send MODE
 */
int main(int argc, char** argv) {
	if (argc >= 2) {
		testRecv();
	} else {
		testSend();
	}
	return 0;
}

